package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class test3 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //读取kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test3");
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));
        //筛选订单和订单明细数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_info"));
        SingleOutputStreamOperator<String> ordercargoDS = stream.filter(x -> JSON.parseObject(x).getString("table").equals("order_cargo"));
        //对流中的数据类型进行转换
        SingleOutputStreamOperator<JSONObject> mapDS = orderInfoDS.map(x -> JSON.parseObject(x));



        //（3）指定流处理环境以及检查点相关的设置，从kafka的tms_ods主题中读取，创建消费者对象，消费数据
        // 封装为流，筛选订单和订单明细数据，对流中的数据类型进行转换 ，按照order_id进行分组，定义侧输出流标签
        // 下单放到主流，支付成功、取消运单、揽收(接单)、发单 转运完成、派送成功、签收放到侧输出流，
        // 从主流中提取侧输出流，将不同流的数据写到kafka的不同主题中（5分）
        //60020
        OutputTag<String> t60020 = new OutputTag<String>("t60020"){};
        //60030
        OutputTag<String> t60030 = new OutputTag<String>("t60030"){};
        //60050
        OutputTag<String> t60050 = new OutputTag<String>("t60050"){};
        //60060
        OutputTag<String> t60060 = new OutputTag<String>("t60060"){};
        //60070
        OutputTag<String> t60070 = new OutputTag<String>("t60070"){};
        //60080
        OutputTag<String> t60080 = new OutputTag<String>("t60080"){};
        //60999
        OutputTag<String> t60999 = new OutputTag<String>("t60999"){};

        SingleOutputStreamOperator<String> t60010DS = mapDS.keyBy(x -> x.getJSONObject("data").getString("id")).process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                JSONObject data = value.getJSONObject("data");
                String status = data.getString("status");
                if ("60010".equals(status)) {
                    out.collect(data.toJSONString());
                } else if ("60020".equals(status)) {
                    ctx.output(t60020, data.toJSONString());
                } else if ("60030".equals(status)) {
                    ctx.output(t60030, data.toJSONString());
                } else if ("60050".equals(status)) {
                    ctx.output(t60050, data.toJSONString());
                } else if ("60060".equals(status)) {
                    ctx.output(t60060, data.toJSONString());
                } else if ("60070".equals(status)) {
                    ctx.output(t60070, data.toJSONString());
                } else if ("60080".equals(status)) {
                    ctx.output(t60080, data.toJSONString());
                } else if ("60999".equals(status)) {
                    ctx.output(t60999, data.toJSONString());
                }
            }
        });
        t60010DS.print();


        FlinkKafkaProducer<String> p60010 = new FlinkKafkaProducer<>("dwd-order-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60020 = new FlinkKafkaProducer<>("dwd-pay-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60030 = new FlinkKafkaProducer<>("dwd-receive-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60050 = new FlinkKafkaProducer<>("dwd-dispatch-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60060 = new FlinkKafkaProducer<>("dwd-bound-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60070 = new FlinkKafkaProducer<>("dwd-deliver-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60080 = new FlinkKafkaProducer<>("dwd-sign-detail", new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> p60999 = new FlinkKafkaProducer<>("dwd-cancel-detail", new SimpleStringSchema(), properties);

        t60010DS.addSink(p60010);
        t60010DS.getSideOutput(t60020).addSink(p60020);
        t60010DS.getSideOutput(t60030).addSink(p60030);
        t60010DS.getSideOutput(t60050).addSink(p60050);
        t60010DS.getSideOutput(t60060).addSink(p60060);
        t60010DS.getSideOutput(t60070).addSink(p60070);
        t60010DS.getSideOutput(t60080).addSink(p60080);
        t60010DS.getSideOutput(t60999).addSink(p60999);

        env.execute();
    }
}
